package com.superiron.scaffold.core.service.mq;


import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.selector.SelectMessageQueueByHash;
import org.apache.rocketmq.common.message.Message;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class RocketProduceService {
	
	public RocketProduceService() {
	}
	
	public SendResult sendMessage(String producerId,  String topic, String msg) throws Exception {
		return this.sendMessage(producerId, topic, "", msg);
	}
	
	public SendResult sendMessage(String producerId,  String topic, String tags, String msg) throws Exception {
		return this.sendMessage(producerId, topic, tags, "", msg);
	}
	
	public SendResult sendMessage(String producerId,  String topic, String tags, String keys, String msg) throws Exception {
		Message message = new Message(topic, tags, keys, msg.getBytes("utf-8"));
		SendResult sendResult = MqUtil.getProducer(producerId).send(message);
		log.info("数据流转rocket：地址：{}，结果：{}", producerId, JSONUtil.toJsonStr(sendResult));
		return sendResult;
	}
	
	public void sendAsyncMessage(String producerId,  String topic, String msg) throws Exception {
		this.sendAsyncMessage(producerId, topic, "", msg);
	}
	
	public void sendAsyncMessage(String producerId,  String topic, String tags, String msg) throws Exception {
		this.sendAsyncMessage(producerId, topic, tags, "", msg);
	}
	
	/**
	 * @param topic
	 * @param tags
	 * @param keys         索引建，空格分隔，快速检索到消息
	 * @param msg
	 *
	 * @throws Exception
	 */
	public void sendAsyncMessage(String producerId,  String topic, String tags, String keys, String msg) throws Exception {
		this.sendAsyncMessage(producerId, topic, tags, keys, msg, getDefaultSendCallBack(producerId));
	}
	
	public void sendAsyncMessage(String producerId,  String topic, String tags, String keys, String msg, SendCallback callback) throws Exception {
		Message message = new Message(topic, tags, keys, msg.getBytes("utf-8"));
		message.isWaitStoreMsgOK();
		MqUtil.getProducer(producerId).send(message, callback);
	}
	
	public SendResult sendDelayMessage(String producerId,  String topic, String msg, int delayLevel) throws Exception {
		Message message = new Message(topic, msg.getBytes("utf-8"));
		message.setDelayTimeLevel(delayLevel);
		return MqUtil.getProducer(producerId).send(message);
	}
	
	public SendResult sendOrderMessage(String producerId,  String topic, String msg, String key) throws Exception {
		Message message = new Message(topic, msg.getBytes("utf-8"));
		return MqUtil.getProducer(producerId).send(message, new SelectMessageQueueByHash(), key);
	}
	
	public SendResult sendOrderDelayMessage(String producerId,  String topic, String msg, int delayLevel, String key) throws Exception {
		Message message = new Message(topic, msg.getBytes("utf-8"));
		message.setDelayTimeLevel(delayLevel);
		return MqUtil.getProducer(producerId).send(message, new SelectMessageQueueByHash(), key);
	}
	
	/**
	 * 默认CallBack函数
	 *
	 * @return
	 */
	private SendCallback getDefaultSendCallBack(String producerId) {
		return new SendCallback() {
			@Override
			public void onSuccess(SendResult sendResult) {
				log.info("---发送MQ成功---：{}，{}", producerId, JSONUtil.toJsonStr(sendResult));
			}
			@Override
			public void onException(Throwable throwable) {
				log.error("---发送MQ失败---"+throwable.getMessage(), throwable.getMessage());
			}
		};
	}
}
